# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
from jms.ods.oms.yl_oms_oms_waybill import jms_ods__yl_oms_oms_waybill

label = f"yl_oms_oms_waybill_new_{int(time())}"
timeout = timedelta(hours=1).seconds
doris_jms_ods__yl_oms_oms_waybill_new = RichSqlSensor(
    task_id='doris_jms_ods__yl_oms_oms_waybill_new',
    pool='broker_load_pool',
    email=['rabie.zhuang@jtexpress.com','yl_bigdata@yl-scm.com'],
    task_concurrency=1,
    conn_id='doris',
    pre_sql=f"""
                 LOAD LABEL jms_ods.{label} (
                     DATA INFILE("hdfs://yl-bg-hdp/dw/hive/jms_ods.db/external/yl_oms_oms_waybill/dt={{{{ execution_date | date_add(-90) |cst_ds }}}}/*")
                     INTO TABLE yl_oms_oms_waybill_new
                     FORMAT AS 'PARQUET'
                        (
                        id                  
                        ,input_time        
                        ,delivery_time        
                        ,pick_network_code 
                        ,waybill_no
                        ,goods_type_id 
                        ,goods_type_code 
                        ,goods_type_name
                        ,package_volume 
                        ,package_total_volume 
                        ,package_charge_weight 
                        ,collect_staff_code 
                        ,collect_staff_name 
                        ,collect_time
                        ,dispatch_staff_code 
                        ,dispatch_staff_name 
                        ,dispatch_time 
                        ,dispatch_network_id 
                        ,dispatch_network_code 
                        ,dispatch_network_name
                        ,sign_network_id  
                        ,sign_network_code 
                        ,sign_network_name 
                        ,sign_time 
                        ,customer_code 
                        ,customer_name 
                        ,sender_province_name
                        ,sender_city_name 
                        ,sender_area_name 
                        ,sender_township 
                        ,sender_detailed_address
                        ,origin_id 
                        ,origin_code
                        ,origin_name 
                        ,pick_network_id
                        ,pick_network_name 
                        ,freight 
                        ,total_freight
                        ,handicraft_fee 
                        ,other_fee 
                        ,receive_pay_fee 
                        ,pick_finance_id 
                        ,pick_finance_code 
                        ,pick_finance_name 
                        ,dispatch_finance_code 
                        ,dispatch_finance_name
                        ,express_type_id 
                        ,express_type_code 
                        ,express_type_name 
                        ,insured 
                        ,insured_amount
                        ,insured_fee
                        ,cod_need 
                        ,cod_money 
                        ,cod_fee 
                        ,settlement_name 
                        ,receiver_name
                        ,receiver_country_name
                        ,receiver_province_name
                        ,receiver_city_name 
                        ,receiver_area_name 
                        ,receiver_township 
                        ,receiver_detailed_address
                        ,destination_id
                        ,destination_code 
                        ,destination_name 
                        ,input_staff_code 
                        ,input_staff_name 
                        ,input_network_code 
                        ,input_network_name 
                        ,coupon_amount 
                        ,is_cod_receive
                        ,settlement_weight
                        ,order_source_code 
                        ,package_collect_weight 
                        ,package_receipt_weight 
                        ,package_arrival_weight 
                        ,package_inside_charge_weight
                        ,is_distribute_scan 
                        ,is_arrival_scan 
                        ,waybill_weight 
                        ,customer_network_code 
                        ,customer_network_name 
                        ,customer_type 
                        ,calculate_fee_desc 
                        ,calculate_fee_code 
                        ,package_center_arrival_weight               
                        )
                        
                 )
                 WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
                 PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_ods WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_ods WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120),
)

doris_jms_ods__yl_oms_oms_waybill_new << [jms_ods__yl_oms_oms_waybill,]

